package avicit.bdp.dds.common.utils;

import avicit.bdp.dds.dispatch.enums.TaskType;
import avicit.bdp.dds.dispatch.task.AbstractParameters;
import avicit.bdp.dds.dispatch.task.algorithm.AlgorithmParameters;
import avicit.bdp.dds.dispatch.task.collect.MetadataCollectParameters;
import avicit.bdp.dds.dispatch.task.conditions.ConditionsParameters;
import avicit.bdp.dds.dispatch.task.datax.DataxParameters;
import avicit.bdp.dds.dispatch.task.dependent.DependentParameters;
import avicit.bdp.dds.dispatch.task.flink.FlinkParameters;
import avicit.bdp.dds.dispatch.task.ft.FTParameters;
import avicit.bdp.dds.dispatch.task.http.HttpParameters;
import avicit.bdp.dds.dispatch.task.java.JavaParameters;
import avicit.bdp.dds.dispatch.task.monitordir.MonitorDirectoryParameters;
import avicit.bdp.dds.dispatch.task.mr.MapreduceParameters;
import avicit.bdp.dds.dispatch.task.procedure.ProcedureParameters;
import avicit.bdp.dds.dispatch.task.python.PythonParameters;
import avicit.bdp.dds.dispatch.task.shell.ShellParameters;
import avicit.bdp.dds.dispatch.task.spark.SparkParameters;
import avicit.bdp.dds.dispatch.task.sparkflow.SparkFlowParameters;
import avicit.bdp.dds.dispatch.task.sql.SqlParameters;
import avicit.bdp.dds.dispatch.task.sqoop.SqoopParameters;
import avicit.bdp.dds.dispatch.task.standard.DataElementEvaluateTaskParameters;
import avicit.bdp.dds.dispatch.task.subprocess.SubProcessParameters;
import avicit.bdp.dds.dispatch.task.table.TableParameters;
import avicit.bdp.core.util.json.JSONUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author admin task parameters utils
 */
public class TaskParametersUtils {

  private static final Logger logger = LoggerFactory.getLogger(TaskParametersUtils.class);

  /**
   * get task parameters
   *
   * @param taskType task type
   * @param parameter parameter
   * @return task parameters
   */
  public static AbstractParameters getParameters(String taskType, String parameter) {
    try {

      switch (EnumUtils.getEnum(TaskType.class, taskType)) {
        case SUB_PROCESS:
          return JSONUtils.parseObject(parameter, SubProcessParameters.class);
        case SHELL:
          return JSONUtils.parseObject(parameter, ShellParameters.class);
        case PROCEDURE:
          return JSONUtils.parseObject(parameter, ProcedureParameters.class);
        case SQL:
          return JSONUtils.parseObject(parameter, SqlParameters.class);
        case MR:
          return JSONUtils.parseObject(parameter, MapreduceParameters.class);
        case SPARK:
          return JSONUtils.parseObject(parameter, SparkParameters.class);
        case PYTHON:
          return JSONUtils.parseObject(parameter, PythonParameters.class);
        case DEPENDENT:
          return JSONUtils.parseObject(parameter, DependentParameters.class);
        case FLINK:
          return JSONUtils.parseObject(parameter, FlinkParameters.class);
        case HTTP:
          return JSONUtils.parseObject(parameter, HttpParameters.class);
        case DATAX:
        case DATAXINCREMENT:
          return JSONUtils.parseObject(parameter, DataxParameters.class);
        case CONDITIONS:
          return JSONUtils.parseObject(parameter, ConditionsParameters.class);
        case SQOOP:
          return JSONUtils.parseObject(parameter, SqoopParameters.class);
        case JAVA:
          return JSONUtils.parseObject(parameter, JavaParameters.class);
        case MODEL:
          return JSONUtils.parseObject(parameter, TableParameters.class);
        case ALGORITHM:
          return JSONUtils.parseObject(parameter, AlgorithmParameters.class);
        case SPARKSQL:
          return JSONUtils.parseObject(parameter, AlgorithmParameters.class);
        case MONITORDIRECTORY:
          return JSONUtils.parseObject(parameter, MonitorDirectoryParameters.class);
        case SPARKFLOW:
          return JSONUtils.parseObject(parameter, SparkFlowParameters.class);
        case METADATA_COLLECT:
          return JSONUtils.parseObject(parameter, MetadataCollectParameters.class);
        case FTTASK:
          return JSONUtils.parseObject(parameter, FTParameters.class);
        case DATA_ELEMENT_EVALUATE:
          return JSONUtils.parseObject(parameter, DataElementEvaluateTaskParameters.class);
        default:
          return null;
      }
    } catch (Exception e) {
      logger.error(e.getMessage(), e);
    }
    return null;
  }
}
